package com.deep.cdc;

import com.deep.bean.ConnectConfig;
import com.deep.bean.TransformConfig;
import com.deep.common.ArgsConfig;
import com.deep.common.ArgsUtil;
import com.deep.common.ProfileConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class SyncRecAct {
    public static void main(String[] args) {
        ArgsConfig argsConfig = ArgsUtil.getMap(args);
        Integer recid = argsConfig.getRecid();

        TransformConfig transformConfig = ProfileConfig.getTransformConfig(argsConfig.getProfile());
        ConnectConfig sourceConfig = transformConfig.getSource();
        ConnectConfig targetConfig = transformConfig.getTarget();


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //创建用于输出的表
        String createInputDDL = "create table source_t_wf_recact " +
                "( " +
                "ACTID  int  , " +
                "PROCCODE  string  , " +
                "PROCID  int  , " +
                "ACTCODE  string  , " +
                "ACTDEFID  int  , " +
                "ACTDEFNAME  string  , " +
                "RECID  int  , " +
                "PARTID  int  , " +
                "PARTNAME  string  , " +
                "UNITID  int  , " +
                "UNITNAME  string  , " +
                "ROLEPARTID  int  , " +
                "ROLEPARTNAME  string  , " +
                "SUPERVISESTATEID  int  , " +
                "SUPERVISESTATENAME  string  , " +
                "ACCREDITSTATEID  int  , " +
                "ACCREDITSTATENAME  string  , " +
                "ACCREDITWORKITEMID  string  , " +
                "CREATETIME  timestamp(0)  , " +
                "DEADLINE  timestamp(0)  , " +
                "WARNINGTIME  timestamp(0)  , " +
                "ENDTIME  timestamp(0)  , " +
                "PREACTID  string  , " +
                "PREPARTID  string  , " +
                "PREPARTNAME  string  , " +
                "PREACTDEFID  string  , " +
                "PREACTDEFNAME  string  , " +
                "PREACTOPINION  string  , " +
                "NEXTACTID  string  , " +
                "NEXTPARTID  string  , " +
                "NEXTPARTNAME  string  , " +
                "NEXTACTDEFID  string  , " +
                "NEXTACTDEFNAME  string  , " +
                "ACTTIMESTATEID  int  , " +
                "ACTLIMITINFO  string  , " +
                "ACTTIMEUNIT  int  , " +
                "ACTLIMIT  double  , " +
                "ACTWARNINGLIMIT  double  , " +
                "ACTUSEDTIME  double  , " +
                "ACTREMAINTIME  double  , " +
                "ACTUSEDTIMECHAR  string  , " +
                "ACTREMAINTIMECHAR  string  , " +
                "REFRESHFLAG  int  , " +
                "REFRESHSTARTTIME  timestamp(0)  , " +
                "REFRESHTIME  timestamp(0)  , " +
                "REFRESHSTOPCOUNT  int  , " +
                "TIMEADDUPFLAG  int  , " +
                "BACKACTID  string  , " +
                "REPLYBACKACTID  string  , " +
                "TRANSWORKITEMID  int  , " +
                "TRANSWORKITEMNAME  string  , " +
                "READFLAG  tinyint  , " +
                "READTIME  timestamp(0)  , " +
                "ROLLBACKENABLED  int  , " +
                "TRANSOPINION  string  , " +
                "ARDINSTDETAILNO  int  , " +
                "ACTUSEDTIMECUR  double  , " +
                "ISTIMINGSTOP  int  , " +
                "ACTUSEDTIMEBEFORESTOP  double  , " +
                "ACTRESTART  timestamp  , " +
                "ACTUSEDTIMECURCHAR  string  , " +
                "OVERTIME  tinyint  , " +
                "ACCEPTTIME  timestamp  , " +
                "ACCEPTPARTID  int  , " +
                "ACCEPTPARTNAME  string , " +
                "  PRIMARY KEY (`ACTID`) NOT ENFORCED   " +
                ")WITH ( " +
                "    'connector' = 'mysql-cdc', " +
                "    'hostname' = '" + sourceConfig.getHostname() + "',  " +
                "    'port' = '" + sourceConfig.getPort() + "',   " +
                "    'username' = '" + sourceConfig.getUsername() + "',  " +
                "    'password' = '" + sourceConfig.getPassword() + "',   " +
                "    'database-name' = '" + sourceConfig.getDatabaseName() + "',  " +
                "    'table-name' = 't_wf_recact'," +
                "   'debezium.skipped.operations'='d' " +
                ") ";
        tableEnv.executeSql(createInputDDL);


        //创建用于输出的表
        String createOutputDDL = "create table target_t_wf_recact " +
                "( " +
                "ACTID  int  , " +
                "PROCCODE  string  , " +
                "PROCID  int  , " +
                "ACTCODE  string  , " +
                "ACTDEFID  int  , " +
                "ACTDEFNAME  string  , " +
                "RECID  int  , " +
                "PARTID  int  , " +
                "PARTNAME  string  , " +
                "UNITID  int  , " +
                "UNITNAME  string  , " +
                "ROLEPARTID  int  , " +
                "ROLEPARTNAME  string  , " +
                "SUPERVISESTATEID  int  , " +
                "SUPERVISESTATENAME  string  , " +
                "ACCREDITSTATEID  int  , " +
                "ACCREDITSTATENAME  string  , " +
                "ACCREDITWORKITEMID  string  , " +
                "CREATETIME  timestamp(0)  , " +
                "DEADLINE  timestamp(0)  , " +
                "WARNINGTIME  timestamp(0)  , " +
                "ENDTIME  timestamp(0)  , " +
                "PREACTID  string  , " +
                "PREPARTID  string  , " +
                "PREPARTNAME  string  , " +
                "PREACTDEFID  string  , " +
                "PREACTDEFNAME  string  , " +
                "PREACTOPINION  string  , " +
                "NEXTACTID  string  , " +
                "NEXTPARTID  string  , " +
                "NEXTPARTNAME  string  , " +
                "NEXTACTDEFID  string  , " +
                "NEXTACTDEFNAME  string  , " +
                "ACTTIMESTATEID  int  , " +
                "ACTLIMITINFO  string  , " +
                "ACTTIMEUNIT  int  , " +
                "ACTLIMIT  double  , " +
                "ACTWARNINGLIMIT  double  , " +
                "ACTUSEDTIME  double  , " +
                "ACTREMAINTIME  double  , " +
                "ACTUSEDTIMECHAR  string  , " +
                "ACTREMAINTIMECHAR  string  , " +
                "REFRESHFLAG  int  , " +
                "REFRESHSTARTTIME  timestamp(0)  , " +
                "REFRESHTIME  timestamp(0)  , " +
                "REFRESHSTOPCOUNT  int  , " +
                "TIMEADDUPFLAG  int  , " +
                "BACKACTID  string  , " +
                "REPLYBACKACTID  string  , " +
                "TRANSWORKITEMID  int  , " +
                "TRANSWORKITEMNAME  string  , " +
                "READFLAG  tinyint  , " +
                "READTIME  timestamp(0)  , " +
                "ROLLBACKENABLED  int  , " +
                "TRANSOPINION  string  , " +
                "ARDINSTDETAILNO  int  , " +
                "ACTUSEDTIMECUR  double  , " +
                "ISTIMINGSTOP  int  , " +
                "ACTUSEDTIMEBEFORESTOP  double  , " +
                "ACTRESTART  timestamp  , " +
                "ACTUSEDTIMECURCHAR  string  , " +
                "OVERTIME  tinyint  , " +
                "ACCEPTTIME  timestamp  , " +
                "ACCEPTPARTID  int  , " +
                "ACCEPTPARTNAME  string , " +
                "  PRIMARY KEY (`ACTID`) NOT ENFORCED  " +
                ")WITH ( " +
                " 'connector' = 'jdbc', " +
                " 'driver' = 'com.mysql.cj.jdbc.Driver', " +
                " 'url' = 'jdbc:mysql://"+ targetConfig.getHostname() + ":" + targetConfig.getPort() + "/" + targetConfig.getDatabaseName() + "?rewriteBatchedStatements=true&useUnicode=true&characterEncoding=utf-8&useSSL=false&serverTimezone=UTC',  " +
                " 'username' = '" + targetConfig.getUsername() + "',  " +
                "  'password' = '" + targetConfig.getPassword() + "',   " +
                "  'table-name' = 't_wf_recact', " +
                "  'connection.max-retry-timeout' = '60s' " +
                ")";
        tableEnv.executeSql(createOutputDDL);


        TableResult tableResult = tableEnv.executeSql("insert into target_t_wf_recact select * from source_t_wf_recact where RECID >  " + recid);

        tableResult.print();

    }
}
